package ${project.packageName}.${project.dataCenterType}.client;


import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import com.alibaba.fastjson.JSONArray;

import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * @author caiqiang
 * @version 2016年7月28日
 * @see HbaseOperateClient
 * @since
 */
@Component(value = "hbaseOperateClient")
public class HbaseOperateClient {

 /**
     * 日志对象
     */
    private static final Logger LOG = LoggerFactory.getLogger(HbaseOperateClient.class);

    /**
     * Description: 单行查询
     * 
     * @param rowkey
     * @param tableName
     * @return Result
     * @throws IOException
     * @see
     */
    public Result getResult(byte[] rowkey, String tableName)
        throws IOException {

        Get get = new Get(rowkey);
        HTableInterface table = HBaseClient.getHConnection().getTable(tableName);
        Result result = table.get(get);

        table.close();
        return result;
    }

    /**
     * Description: 单元格查询
     * 
     * @param rowkey
     * @param tableName
     * @param family
     * @param qualifier
     * @return Result
     * @throws IOException
     * @see
     */
    public Result getResult(byte[] rowkey, String tableName, byte[] family, byte[] qualifier)
        throws IOException {

        Get get = new Get(rowkey);
        get.addColumn(family, qualifier);
        HTableInterface table = HBaseClient.getHConnection().getTable(tableName);
        Result result = table.get(get);

        table.close();
        return result;
    }

    /**
     * Description:范围查询
     * 
     * @param startRowkey
     * @param endRowkey
     * @param tableName
     * @return ResultScanner
     * @throws IOException
     *             ResultScanner
     * @see
     */
    public ResultScanner getResultScanner(byte[] startRowkey, byte[] endRowkey, String tableName)
        throws IOException {

        Scan scan = new Scan(startRowkey, endRowkey);
        scan.setSmall(true);
        HTableInterface table = HBaseClient.getHConnection().getTable(tableName);
        ResultScanner rs = table.getScanner(scan);

        table.close();
        return rs;
    }
  /**
     * Description: 获取hbase中所有的表名
     * 
     * @return List<TableName>
     * @see
     */
    @SuppressWarnings("resource")
    public static List<TableName> getTableNames() {
        try {

            HBaseAdmin hBaseAdmin = new HBaseAdmin(HBaseClient.getConfiguration());
            List<TableName> tableNames = Arrays.asList(hBaseAdmin.listTableNames());
            return tableNames;
        }
        catch (MasterNotRunningException e) {
            LOG.info("hbase master not running");
            e.printStackTrace();
            return new ArrayList<TableName>();
        }
        catch (ZooKeeperConnectionException e) {
            LOG.info("zookeeper does not connect");
            e.printStackTrace();
            return new ArrayList<TableName>();
        }
        catch (IOException e) {
            e.printStackTrace();
            return new ArrayList<TableName>();
        }
    }
   /**
     * Description:实现HBase 伪分页
     * 
     * @param resultScanner
     * @return Map<Integer,Result>
     * @see
     */
    public static List<Object> getPageResults(JSONArray returnArrayResult, int pageSize,
                                              int pageNumber) {

        int rowIndex = 0;
        List<Object> objects = new ArrayList<Object>();

        int startIndex = (pageNumber - 1) * pageSize;
        int endIndex = startIndex + pageSize;

        for (Object object : returnArrayResult) {

            if (startIndex <= rowIndex && rowIndex < endIndex) {
                objects.add(object);
            }
            rowIndex++ ;
        }
        return objects;
    }
}
